package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author zhangsan
 * @Date 2022/10/19 11:50
 * @Description //TODO dwd层 考试域 试卷问题考试相关信息
 */
public class DwdTestPaperQueInfo {
    public static void main(String[] args) throws Exception{
        //TODO 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //TODO 2.检查点设置
        //TODO 3.从kafka 的topic_db中读取数据
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("test_paper_que_info_group"));
        //TODO 4.过滤出 考试问题表中的数据 test_exam_question
        Table paperQueTable = tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['exam_id'] exam_id,\n" +
                "`data`['paper_id'] paper_id,\n" +
                "`data`['question_id'] question_id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['is_correct'] is_correct,\n" +
                "`data`['score'] score,\n" +
                "`data`['create_time'] ts\n" +
                "from topic_db\n" +
                "where `table` = 'test_exam_question'\n" +
                "and `data`['deleted'] = '0'");
        tableEnv.createTemporaryView("paper_que_temp",paperQueTable);
        //TODO 5.将结果写到kafka对应的主题中
        tableEnv.executeSql("CREATE TABLE dwd_test_paper_que_exam (\n" +
                "  id string,\n" +
                "  exam_id string,\n" +
                "  paper_id string,\n" +
                "  question_id  string,\n" +
                "  user_id  string,\n" +
                "  is_correct string,\n" +
                "  score string,\n" +
                "  ts string,\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ") " + MyKafkaUtil.getUpsertKafkaDDL("dwd_test_paper_que_exam"));

        tableEnv.executeSql("select * from paper_que_temp").print();
        tableEnv.executeSql("insert into dwd_test_paper_que_exam select * from paper_que_temp");


    }
}
